#include "redis.hh"
#include "../util/os_util.hh"
#include "../util/string_util.hh"
#include "../../object_pool.hpp"
#include "../logger/writer.hh"
#include <random>
#include <ctime>

#ifdef WIN32
#include <Winsock2.h>
#else
#include <sys/time.h>
#endif

#ifdef KRATOS_COMPILER_LOG_OFF

#define REDIS_LOG_ERROR(a, b)

#define REDIS_LOG_EXCEPTION(e, a, b)

#define REDIS_PARAM_FAIL_RETURN_FALSE(a)\
    return false;

#define REDIS_PARAM_FAIL_RETURN(a, b)\
    return b;

#else

#define REDIS_LOG_ERROR(a, b)\
    logwcom("Redis") << a << b << logend;

#define REDIS_LOG_EXCEPTION(e, a, b)\
    logwcom("Redis") << "Exception:" << e.what() <<","<< a << b << logend;

#define REDIS_PARAM_FAIL_RETURN_FALSE(a)\
    if (a) {\
        logwcom("Redis") << "[" << __FILE__ << ":" << __LINE__ << "]:" << #a << ", check failed" << logend;\
        return false;\
    }

#define REDIS_PARAM_FAIL_RETURN(a, b)\
    if (a) {\
        logwcom("Redis") << "[" << __FILE__ << ":" << __LINE__ << "]:" << #a << ", check failed" << logend;\
        return b;\
    }

#endif // KRATOS_COMPILER_LOG_OFF

using namespace kratos::corelib;

namespace kratos { namespace redis {

Redis::Redis() {
}

Redis::~Redis() {
    for (auto it : hostMap_) {
        for (auto conn : it.second) {
            delete conn;
        }
    }
    hostMap_.clear();
}

void Redis::update(std::time_t now) {
    for (auto it : hostMap_) {
        for (auto conn : it.second) {
            conn->update(now);
        }
    }
}

bool Redis::addHost(const std::string & name, const std::string & host, const std::string & password, std::size_t maxPoolSize) {
    REDIS_PARAM_FAIL_RETURN_FALSE(name.empty() || host.empty());
    if (!checkHost(host)) {
        return false;
    }
    if (!maxPoolSize) {
        maxPoolSize = 4;
    }
    for (std::size_t i = 0; i < maxPoolSize; i++) {
        auto conn = new RedisConn(host, password);
        if (!conn->start()) {
            REDIS_LOG_ERROR("Start redis connection failed, host:", host);
        }
        hostMap_[name].push_back(conn);
    }
    return true;
}

bool Redis::removeHost(const std::string & name) {
    REDIS_PARAM_FAIL_RETURN_FALSE(name.empty());
    auto it = hostMap_.find(name);
    if (it == hostMap_.end()) {
        REDIS_LOG_ERROR("Redis host not found, name:", name);
        return false;
    }
    for (auto conn : it->second) {
        if (conn->stop()) {
            delete conn;
        } else {
            REDIS_LOG_ERROR("Stop redis connection", " failed");
        }
    }
    hostMap_.erase(it);
    return true;
}

RedisConn & Redis::findConn(const std::string & hostName, HostFindPolicy policy) {
    (void)policy;
    static RedisConn defaultConn;
    if (hostName.empty()) {
        REDIS_LOG_ERROR("Redis connection not found for ", hostName);
        return defaultConn;
    }
    static std::default_random_engine e;
    auto& hostList = hostMap_[hostName];
    if (hostList.empty()) {
        REDIS_LOG_ERROR("Redis connection not found for ", hostName);
        return defaultConn;
    }
    return *hostList[e() % hostList.size()];
}

bool Redis::checkHost(const std::string & host) {
    std::vector<std::string> result;
    corelib::StringUtil::split(host, ":", result);
    REDIS_PARAM_FAIL_RETURN_FALSE(result.size() > 2);
    return true;
}


RedisConn::RedisConn(const std::string & host, const std::string & password)
    : host_(host), password_(password), ctx_(nullptr), run_(false) {
}

RedisConn::RedisConn()
    : ctx_(nullptr), run_(false) {
}

RedisConn::~RedisConn() {
    stop();    
}

void RedisConn::update(std::time_t now) {
    auto timerID = timerQueue_.checkExpired();
    while (timerID != corelib::INVALID_TIMER_ID) {
        auto it = timerMap_.find(timerID);
        if (it != timerMap_.end()) {
            auto methodsIt = cbMap_.find(it->second);
            if (methodsIt != cbMap_.end()) {
                auto& methods = methodsIt->second;
                if (methods.timeoutMethod) {
                    methods.timeoutMethod(*methods.result, *this);
                } else {
                    logwcom("Redis") << "Command timeout, taskID:" << methods.result->getTaskID()
                        << ", Command:" << methods.result->getCommand() << corelib::logend;
                }
            }
            timerMap_.erase(it);
        }
        timerID = timerQueue_.checkExpired();
    }
    RedisResult* result = nullptr;
    while (output_.try_dequeue(result)) {
        auto it = cbMap_.find(result->getSequence());
        if (it == cbMap_.end()) {
            corelib::dispose(result);
            continue;
        }
        if (*result) {
            if (it->second.finishMethod) {
                it->second.finishMethod(*result, *this);
            }
        } else {
            if (it->second.errorMethod) {
                it->second.errorMethod(*result, *this);
            } else {
                logwcom("Redis") << "Command error, taskID:" << result->getTaskID() << ", error:"
                    << result->getErrorString() << ", Command:" << result->getCommand() << corelib::logend;
            }
        }
        cbMap_.erase(it);
        corelib::dispose(result);
    }
}

bool RedisConn::start() {
    REDIS_PARAM_FAIL_RETURN_FALSE(run_);
    run_ = true;
    timerQueue_.start();
    routine_.reset(new std::thread([&, this]() {
        while (run_) {
            if (!ctx_) {
                ctx_ = connect();
            }
            if (!ctx_) {
                continue;
            }
            RedisResult* result;
            if (input_.try_dequeue(result)) {
                auto reply = doRedisCommand(result->getCommand());
                result->reset(reply);
                result->setErrorString(lastError_);
                if (!output_.try_enqueue(result)) {
                    corelib::dispose(result);
                }
            } else {
                std::this_thread::sleep_for(std::chrono::milliseconds(1));
            }
        }
    }));
    return true;
}

bool RedisConn::stop() {
    run_ = false;
    if (routine_) {
        routine_->join();
        routine_.reset(nullptr);
    }
    timerQueue_.stop();
    RedisResult* result = nullptr;
    while (input_.try_dequeue(result)) {
        corelib::dispose(result);
    }
    while (output_.try_dequeue(result)) {
        corelib::dispose(result);
    }
    if (ctx_) {
        redisFree(ctx_);
    }
    ctx_ = nullptr;
    return true;
}

const std::string & RedisConn::getLastError() {
    return lastError_;
}

RedisConn::operator bool() {
    return (ctx_ != nullptr) || !run_;
}

redisReply* RedisConn::doRedisCommand(const std::string& command) {
    if (!ctx_) {
        ctx_ = connect();
        if (!ctx_) {
            lastError_ = "Redis not connected";
            return nullptr;
        }
    }
    auto reply = reinterpret_cast<redisReply*>(redisCommand(ctx_, command.c_str()));
    if (reply == nullptr) {
        switch (ctx_->err) {
        case REDIS_ERR_IO:
        case REDIS_ERR_EOF:
        case REDIS_ERR_PROTOCOL:
        case REDIS_ERR_OTHER:
            lastError_ = ctx_->errstr;            
            break;
        }
    }
    if (reply) {
        // reset reply object manually
        ctx_->reader->reply = nullptr;
    }
    return reply;
}

bool RedisConn::doCommand(const std::string & command, std::uint32_t taskID, std::uint64_t userData, std::time_t timeout,
    TaskFinishMethod finishMethod, TaskTimeoutMethod timeoutMethod, TaskErrorMethod errorMethod) {
    static std::uint32_t sequence = 0;
    REDIS_PARAM_FAIL_RETURN_FALSE(command.empty() || !taskID);
    sequence += 1;
    auto result = corelib::allocate<RedisResult>(taskID, userData, sequence);
    REDIS_PARAM_FAIL_RETURN_FALSE(!result);
    result->setCommand(command);
    corelib::TimerID timerID = corelib::INVALID_TIMER_ID;
    if (timeout) {
        timerID = timerQueue_.startTimerOnce(timeout, 0);
        if (timerID == corelib::INVALID_TIMER_ID) {
            corelib::dispose(result);
            return false;
        }
    }
    if (!input_.try_enqueue(result)) {
        corelib::dispose(result);
        return false;
    }
    if (timeout) {
        timerMap_[timerID] = sequence;
    }
    cbMap_[sequence] = { finishMethod, timeoutMethod, errorMethod, timerID, result };
    return true;
}

redisContext * RedisConn::connect() {
    std::vector<std::string> result;
    corelib::StringUtil::split(host_, ":", result);
    int32_t port = 6379;
    std::string host;
    REDIS_PARAM_FAIL_RETURN(result.empty(), nullptr);
    REDIS_PARAM_FAIL_RETURN(result.size() > 2, nullptr);
    host = result[0];
    if (result.size() == 2) {
        port = std::stoi(result[1]);
    }    
    std::string realIp = corelib::OS::getHostIP(host);
    if (realIp.empty()) {
        lastError_ = "Invalid host:" + host_;
        REDIS_LOG_ERROR("Connect to redis server failed:", host);
        return nullptr;
    }
    struct timeval tv;
    tv = { 1, 0 }; // re-connect timeout
    auto ctx = redisConnectWithTimeout(realIp.c_str(), port, tv);
    if (ctx == nullptr) {
        return nullptr;
    } else {
        if (ctx->err) {
            lastError_ = ctx->errstr;
            redisFree(ctx);
            ctx = nullptr;
            return nullptr;
        }
    }
    if (password_.empty()) {
        return ctx;
    }
    auto reply = doRedisCommand("AUTH " + password_);
    if (!reply || (reply->type == REDIS_REPLY_ERROR)) {
        redisFree(ctx);
        ctx = nullptr;
    }
    if (reply) {
        freeReplyObject(reply);
    }
    return ctx;
}

redis::RedisResult::RedisResult(std::uint32_t taskID, std::uint64_t userData, std::uint32_t sequence)
    : reply_(nullptr), taskID_(taskID), userData_(userData), sequence_(sequence) {
}

redis::RedisResult::~RedisResult() {
    if (reply_) {
        freeReplyObject(reply_);
    }
    reply_ = nullptr;
}

void RedisResult::setCommand(const std::string & redisCommand) {
    redisCommand_ = redisCommand;
}

const std::string & RedisResult::getCommand() {
    return redisCommand_;
}

void redis::RedisResult::reset(redisReply * reply) {
    if (reply_) {
        freeReplyObject(reply_);
    }
    reply_ = reply;
}

redisReply * redis::RedisResult::getReply() {
    return reply_;
}

std::uint32_t RedisResult::getTaskID() {
    return taskID_;
}

std::uint64_t RedisResult::getUserData() {
    return userData_;
}

const std::string & RedisResult::getErrorString() const {
    return errorString_;
}

void RedisResult::setErrorString(const std::string & errorString) {
    errorString_ = errorString;
}

std::uint32_t RedisResult::getSequence() {
    return sequence_;
}

RedisResult::operator bool() {
    return (reply_ != nullptr);
}

}}
